package org.carl.dual;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
/**
 * 非阻塞并发线程任务执行
 * @author Carl
 *
 */
import java.util.logging.Logger;
public class ConcurrentThread {
	private static final Logger LOG=Logger.getGlobal();
	private static final int SIZE = Runtime.getRuntime().availableProcessors();
	private static final ExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(SIZE * 100);
	// 信号量
	private Semaphore semaphore;
	{
		init();
	}

	public void init() {
		semaphore = new Semaphore(SIZE, false);// 初始化对象时操作
	}

	public CopyOnWriteArrayList<Future<?>> submit(ConcurrentLinkedQueue<Callable<?>> queue) {
		if (!queue.isEmpty()) {
			CopyOnWriteArrayList<Future<?>> res_list = new CopyOnWriteArrayList<>();
			Callable<?> callable = null;
			while (!queue.isEmpty()) {
				// 从信号量尝试获取一个许可，如果无可用许可，直接返回false，不会阻塞
				if (semaphore.tryAcquire()) {
					callable = queue.poll();
					LOG.info(Thread.currentThread().getName()+" submit task.."+callable.toString());
					res_list.add(EXECUTOR_SERVICE.submit(callable));
					// 释放一个许可，别忘了在finally中使用，注意：多次调用该方法，会使信号量的许可数增加，达到动态扩展的效果，如：初始permits
					// 为1， 调用了两次release，最大许可会改变为2
					semaphore.release();
				} else {
					// 继续非阻塞执行
					continue;
				}
			}
			EXECUTOR_SERVICE.shutdown();
			return res_list;
		}
		return null;
	}
}
